use std::fmt::Debug;
use std::time::Duration;

use futures::channel::mpsc;
use serde::Serialize;
use thiserror::Error;

use crate::actors::{
    actor_center::{self, ActorCenter},
    router::{NatsMsg, NetMsg, NetRpcMsg, NetRpcRouterMsg, Router, RouterContext},
    shutdown, SendMsg, ShutdownActor, TcpStreamActor,
};
use crate::prelude::*;
use crate::Result;

#[async_trait::async_trait]
pub trait ContextExt<A> {
    async fn get_router(&self) -> Result<RouterContext<A>>;
    async fn register_center(&self, name: &str) -> Result<()>
    where
        A: Actor;
    async fn unregister_center(&self) -> Result<()>
    where
        A: Actor;
    async fn register_shutdown(&self) -> Result<()>
    where
        A: Actor;
    async fn unregister_shutdown(&self) -> Result<()>
    where
        A: Actor;
}

#[async_trait::async_trait]
impl<A> ContextExt<A> for Context<A> {
    async fn get_router(&self) -> Result<RouterContext<A>> {
        let router_addr = Router::from_registry().await?;
        let routee_addr = self.address();
        Ok(RouterContext {
            router_addr,
            routee_addr,
        })
    }

    async fn register_center(&self, name: &str) -> Result<()>
    where
        A: Actor,
    {
        let center_addr = ActorCenter::from_registry().await?;
        let addr = self.address();
        center_addr.send(actor_center::Register {
            name: name.into(),
            addr,
        })
    }

    async fn unregister_center(&self) -> Result<()>
    where
        A: Actor,
    {
        let center_addr = ActorCenter::from_registry().await?;
        let addr = self.address();
        center_addr.send(actor_center::Unregister(addr.actor_id()))
    }

    async fn register_shutdown(&self) -> Result<()>
    where
        A: Actor,
    {
        let shutdown_addr = ShutdownActor::from_registry().await?;
        let addr = self.address();
        shutdown_addr.send(shutdown::Register(addr))
    }

    async fn unregister_shutdown(&self) -> Result<()>
    where
        A: Actor,
    {
        let shutdown_addr = ShutdownActor::from_registry().await?;
        shutdown_addr.send(shutdown::UnRegister(self.actor_id()))
    }
}

pub trait RouterAddrExt {
    fn route<T>(&self, source: Addr<TcpStreamActor>, msg: T) -> Result<()>
    where
        T: Message<Result = ()> + HoneybeeMessage;

    fn route_rpc<T, R>(&self, source: Addr<TcpStreamActor>, session: u32, msg: T) -> Result<()>
    where
        T: Message<Result = R> + HoneybeeMessage,
        R: RpcResult;

    fn route_nats<T>(&self, reply: String, msg: T) -> Result<()>
    where
        T: Message<Result = ()> + HoneybeeMessage;
}

impl RouterAddrExt for Addr<Router> {
    fn route<T>(&self, source: Addr<TcpStreamActor>, msg: T) -> Result<()>
    where
        T: Message<Result = ()> + HoneybeeMessage,
    {
        self.send(NetMsg { source, msg })
    }

    fn route_rpc<T, R>(&self, source: Addr<TcpStreamActor>, session: u32, msg: T) -> Result<()>
    where
        T: Message<Result = R> + HoneybeeMessage,
        R: RpcResult,
    {
        self.send(NetRpcRouterMsg {
            source,
            session,
            msg,
        })
    }

    fn route_nats<T>(&self, reply: String, msg: T) -> Result<()>
    where
        T: Message<Result = ()> + HoneybeeMessage,
    {
        self.send(NatsMsg { reply, msg })
    }
}

#[async_trait::async_trait]
pub trait RemoteAddrExt {
    fn send_remote<T>(&self, msg: T) -> Result<()>
    where
        T: Message<Result = ()> + HoneybeeMessage;

    async fn call_remote<T, R>(&self, msg: T) -> std::result::Result<T::Result, RpcResponseError>
    where
        T: Message<Result = R> + HoneybeeMessage,
        R: RpcResult;

    async fn call_remote_timeout<T, R>(
        &self,
        msg: T,
        dur: Duration,
    ) -> std::result::Result<T::Result, RpcResponseError>
    where
        T: Message<Result = R> + HoneybeeMessage,
        R: RpcResult;

    fn async_call_remote<A, T, R>(&self, source: Addr<A>, msg: T) -> Result<()>
    where
        A: Actor + AnyHandler<R>,
        T: Message<Result = R> + HoneybeeMessage,
        R: RpcResult;
}

#[async_trait::async_trait]
pub trait RemotesAddrExt {
    fn send_remote_random<T>(&self, msg: T) -> Result<()>
    where
        T: Message<Result = ()> + HoneybeeMessage;

    async fn call_remote_random<T, R>(
        &self,
        msg: T,
    ) -> std::result::Result<T::Result, RpcResponseError>
    where
        T: Message<Result = R> + HoneybeeMessage,
        R: RpcResult;

    fn async_call_remote_random<A, T, R>(&self, source: Addr<A>, msg: T) -> Result<()>
    where
        A: Actor + AnyHandler<R>,
        T: Message<Result = R> + HoneybeeMessage,
        R: RpcResult;

    fn send_remote_specific<T>(&self, remote_name: String, msg: T) -> Result<()>
    where
        T: Message<Result = ()> + HoneybeeMessage;

    async fn call_remote_specific<T, R>(
        &self,
        remote_name: String,
        msg: T,
    ) -> std::result::Result<T::Result, RpcResponseError>
    where
        T: Message<Result = R> + HoneybeeMessage,
        R: RpcResult;

    fn async_call_remote_specific<A, T, R>(
        &self,
        remote_name: String,
        source: Addr<A>,
        msg: T,
    ) -> Result<()>
    where
        A: Actor + AnyHandler<R>,
        T: Message<Result = R> + HoneybeeMessage,
        R: RpcResult;

    fn send_remote_hash<T>(&self, key: usize, msg: T) -> Result<()>
    where
        T: Message<Result = ()> + HoneybeeMessage;

    async fn call_remote_hash<T, R>(
        &self,
        key: usize,
        msg: T,
    ) -> std::result::Result<T::Result, RpcResponseError>
    where
        T: Message<Result = R> + HoneybeeMessage,
        R: RpcResult;

    fn async_call_remote_hash<A, T, R>(&self, key: usize, source: Addr<A>, msg: T) -> Result<()>
    where
        A: Actor + AnyHandler<R>,
        T: Message<Result = R> + HoneybeeMessage,
        R: RpcResult;

    fn send_remote_all<T>(&self, msg: T) -> Result<()>
    where
        T: Message<Result = ()> + HoneybeeMessage + Clone;
}

pub trait UserSender {
    fn send_to_user<T>(&self, uid: i64, msg: T)
    where
        T: 'static + prost::Message + MessageId;
}

pub trait AllUserSender {
    fn send_to_all_user<T>(&self, msg: T)
    where
        T: 'static + prost::Message + MessageId;
}

pub trait AppUserSender {
    fn send_to_app_user<T>(&self, uid: i64, msg: T)
    where
        T: Message<Result = ()> + HoneybeeMessage;
}

pub trait AllAppUserSender {
    fn send_to_all_app_user<T>(&self, msg: T)
    where
        T: Message<Result = ()> + HoneybeeMessage;
}

pub trait UmsgSender {
    fn reliable_send_to_user<T>(&self, uid: i64, msg_type: umsg_db::UmsgType, msg: T)
    where
        T: Serialize + Debug;
}

pub trait GmsgSender {
    fn reliable_send_to_all_user<T>(
        &self,
        msg_type: gmsg_db::GmsgType,
        msg: T,
        expire: Option<i64>,
    ) where
        T: Serialize + Debug;
}

pub type RpcCallerFn<T> = Box<dyn Fn(Addr<TcpStreamActor>, u32, T) + 'static + Send>;

pub struct RpcCaller<T: Message + HoneybeeMessage>(pub RpcCallerFn<T>);

impl<T> RpcCaller<T>
where
    T: Message + HoneybeeMessage,
{
    pub fn rpc_call(&self, source: Addr<TcpStreamActor>, session: u32, msg: T) {
        self.0(source, session, msg)
    }
}

#[derive(Error, Debug)]
pub struct RpcPending;

impl std::fmt::Display for RpcPending {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "RpcPending")
    }
}

pub trait RpcAddrExt<A> {
    fn rpc_caller<T>(&self) -> RpcCaller<T>
    where
        T: Message + HoneybeeMessage,
        A: Handler<NetRpcMsg<T>>,
        <T as Message>::Result: RpcResult;
}

impl<A> RpcAddrExt<A> for Addr<A> {
    fn rpc_caller<T>(&self) -> RpcCaller<T>
    where
        T: Message + HoneybeeMessage,
        A: Handler<NetRpcMsg<T>>,
        <T as Message>::Result: RpcResult,
    {
        let tx = self.tx.clone();
        RpcCaller(Box::new(move |source, session, msg| {
            let source_ = source.clone();
            if let Err(err) = mpsc::UnboundedSender::clone(&*tx).start_send(ActorEvent::Exec(
                Box::new(move |actor, ctx| {
                    Box::pin(async move {
                        let res = Handler::handle(
                            actor,
                            ctx,
                            NetRpcMsg {
                                session,
                                source: source_.clone(),
                                msg,
                            },
                        )
                        .await;
                        if let Err(ref err) = res {
                            if err.is::<RpcPending>() {
                                return;
                            }
                        }
                        let _ = source_.send(SendMsg {
                            session,
                            msg: RpcResponse(
                                res.map_err(|err| RpcResponseError::RpcError(err.to_string())),
                            ),
                        });
                    })
                }),
            )) {
                error!("err:{}", err);
                let _ = source.send(SendMsg {
                    session,
                    msg: RpcResponse::<T::Result>(Err(RpcResponseError::RpcError(err.to_string()))),
                });
            }
        }))
    }
}
